package com.li.flink.datastream

import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object RichMapOp {
  def main(args: Array[String]): Unit = {
    //获取运行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //连接socket
    val text = env.socketTextStream("bigdata04", 9001)

    //自定义mapFunction
    val mapFunc = new MapFunction[String, String] {
      override def map(t: String) = {
        t + 1
      }
    }
    //处理数据
    //    text.flatMap(_.split(" ")).map(mapFunc).print();

    //RichMapFunction 可以自定义生命周期
    val richMapFunction = new RichMapFunction[String, String] {
      override def map(in: String): String = {
        println("-----map-----")
        in
      }

      //open & close 每个线程调用一次。open中可以执行开启数据库连接等操作
      override def open(parameters: Configuration): Unit = {
        println("-----open-----")
        super.open(parameters)
      }

      override def close(): Unit = {
        println("-----close-----")
        super.close()
      }
    }
    //可以在外部设置并行度
    text.flatMap(_.split(" ")).map(richMapFunction).setParallelism(3).print();

    env.execute("RichMapOp")
  }

}